package com.cloud.rou.im.client.msg;


import com.cloud.rou.im.client.statemachine.MsgEvent;
import com.cloud.rou.im.client.statemachine.StateMachine;
import com.cloud.rou.im.client.statemachine.StateMachineContext;
import com.cloud.rou.im.core.util.SingletonUtil;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * <p></p>消息队列执行器
 *
 * @author GaoShuangWen
 */
public class CheckAckProcessor {

    private final ExecutorService subExecutor;

    private final DelayQueue<AckDelay<StateMachineContext>> delayQueue;
    private Map<Long, AckDelay<StateMachineContext>> cacheMap;

    private CheckAckProcessor() {

        delayQueue = new DelayQueue<>();
        cacheMap = new ConcurrentHashMap<>();
        this.subExecutor = Executors.newFixedThreadPool(1, r -> {
            Thread thread = new Thread(r);
            thread.setDaemon(true);
            thread.setName("com.cloud.rou.im.client.msg.MessageQueueExecutor");
            return thread;
        });
        start();
    }

    public static CheckAckProcessor instance() {
        return SingletonUtil.get(CheckAckProcessor.class);
    }

    class SubWork implements Runnable {
        @Override
        public void run() {
            while (true) {
                try {
                    AckDelay<StateMachineContext> ackDelay = delayQueue.take();
                    StateMachineContext data = ackDelay.getData();
                    StateMachine stateMachine = data.getStateMachine();
                    if (ackDelay.getData().getRetryTime().decrementAndGet() == 1) {
                        System.out.println("次数已经使用完:");
                        stateMachine.fire(MsgEvent.ERROR);
                    } else {
                        stateMachine.fire(MsgEvent.DELAY);
                        stateMachine.fire(MsgEvent.SEND);
                    }
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }

        }
    }

    public void add(StateMachineContext context) {
        AckDelay<StateMachineContext> e = new AckDelay<>(context, 3000L,3);
        delayQueue.offer(e);
        cacheMap.put(context.getMsg().getMsgId(), e);
    }

    public void start() {
        subExecutor.submit(new SubWork());
    }

    public void remove(StateMachineContext context) {
        Long msgId = context.getMsg().getMsgId();
        AckDelay<StateMachineContext> stateMachineContextAckDelay = cacheMap.get(msgId);
        delayQueue.remove(stateMachineContextAckDelay);
        cacheMap.remove(msgId);

    }
}
